package com.nulldev.util.concurrency.threadIt.v4.executors.cfs;

import java.util.List;
import java.util.Map;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import com.nulldev.util.VariableAPI.ArrayUtils;
import com.nulldev.util.concurrency.threadIt.v4.emapi.extensions.ExecutorFeatureExtensions;
import com.nulldev.util.concurrency.threadIt.v4.emapi.extensions.ExtendedExecutor;
import com.nulldev.util.concurrency.threadIt.v4.emapi.uft.Executable;
import com.nulldev.util.concurrency.threadIt.v4.utils.impls.TEConditions;
import com.nulldev.util.logging.Logger;
import com.nulldev.util.logging.LoggerFactory;
import com.nulldev.util.logging.stubs.log4j.MDC;

public final class CompletelyFairExecutorService extends ThreadPoolExecutor implements FairExecutorService, ExtendedExecutor {

	/*
	 * No, not _that_ CFS...
	 */

	private static final Logger LOG = LoggerFactory.getLogger(CompletelyFairExecutorService.class);

	private final int corePoolSize;
	private final int maxPoolSize;
	private final int leasesPerThread;
	private final int maxLeasesPerThread;
	private final long keepAliveTime;
	private final TimeUnit timeUnit;
	private final CompletelyFairWorkQueue workQueue;
	private final ThreadFactory threadFactory;

	public static Builder builder() {
		return new Builder();
	}

	private CompletelyFairExecutorService(final int corePoolSize, final int maxPoolSize, final int leasesPerThread, final int maxLeasesPerThread,
			final long keepAliveTime, final TimeUnit timeUnit, final ThreadFactory threadFactory) {

		this(corePoolSize, maxPoolSize, leasesPerThread, maxLeasesPerThread, keepAliveTime, timeUnit,
				new CompletelyFairWorkQueue(leasesPerThread, maxLeasesPerThread), threadFactory);
	}

	private CompletelyFairExecutorService(final int corePoolSize, final int maxPoolSize, final int leasesPerThread, final int maxLeasesPerThread,
			final long keepAliveTime, final TimeUnit timeUnit, final CompletelyFairWorkQueue workQueue, final ThreadFactory threadFactory) {

		super(corePoolSize, maxPoolSize, keepAliveTime, timeUnit, workQueue, threadFactory);

		this.corePoolSize = corePoolSize;
		this.maxPoolSize = maxPoolSize;
		this.leasesPerThread = leasesPerThread;
		this.maxLeasesPerThread = maxLeasesPerThread;
		this.keepAliveTime = keepAliveTime;
		this.timeUnit = timeUnit;
		this.workQueue = workQueue;
		this.threadFactory = threadFactory;

		init();
	}

	public void init() {
		prestartAllCoreThreads();
	}

	@Override
	public void allowCoreThreadTimeOut(final boolean value) {
		if (value) {
			throw new UnsupportedOperationException("Core thread time out is not allowable");
		}
	}

	public void shutdownAndAwaitTermination() {
		shutdown(); // Disable new tasks from being submitted
		try {
			// Wait a while for existing tasks to terminate
			if (!awaitTermination(60, TimeUnit.SECONDS)) {
				shutdownNow(); // Cancel currently executing tasks
				// Wait a while for tasks to respond to being cancelled
				if (!awaitTermination(60, TimeUnit.SECONDS))
					LOG.error("Failed to shut down in a timely manner");
			}
		} catch (InterruptedException ie) {
			// (Re-)Cancel if current thread also interrupted
			shutdownNow();
			// Preserve interrupt status
			Thread.currentThread().interrupt();
		}
	}

	@Override
	public void newContext() {
		workQueue.newContext();
	}

	@Override
	public void clearContext() {
		workQueue.clearContext();
	}

	@Override
	public long getCurrentContextId() {
		return workQueue.getCurrentContextId();
	}

	@Override
	public FairExecutorService wrapperWithCurrentContext() {
		return new View(workQueue.wrapperWithCurrentContext());
	}

	@Override
	public FairExecutorService wrapperWithNewContext() {
		return new View(workQueue.wrapperWithNewContext());
	}

	@Override
	public CompletableFuture<?> submit(final Runnable task) {
		return FairExecutorService.super.submit(task);
	}

	@Override
	public <T> CompletableFuture<T> submit(final Runnable task, final T result) {
		return FairExecutorService.super.submit(task, result);
	}

	@Override
	public <T> CompletableFuture<T> submit(final Callable<T> task) {
		return FairExecutorService.super.submit(task);
	}

	@Override
	public void execute(final Runnable command) {
		super.execute(wrapCommand(command));
	}

	protected Runnable wrapCommand(final Runnable command) {
		final Map<String, String> mdc = MDC.getCopyOfContextMap();
		if (mdc == null || mdc.isEmpty()) {
			// MDC is empty, no need to wrap the command
			return command;
		}

		return () -> {
			MDC.setContextMap(mdc);
			try {
				command.run();
			} finally {
				MDC.clear();
			}
		};
	}

	private class View extends AbstractExecutorService implements FairExecutorService {

		private final FairWorkQueue queue;

		private View(final FairWorkQueue queue) {
			this.queue = queue;
		}

		@Override
		public void shutdown() {
			throw new UnsupportedOperationException("CompletelyFairExecutorService.View is write only");
		}

		@Override
		public List<Runnable> shutdownNow() {
			throw new UnsupportedOperationException("CompletelyFairExecutorService.View is write only");
		}

		@Override
		public boolean isShutdown() {
			return CompletelyFairExecutorService.this.isShutdown();
		}

		@Override
		public boolean isTerminated() {
			return CompletelyFairExecutorService.this.isTerminated();
		}

		@Override
		public boolean awaitTermination(final long timeout, final TimeUnit unit) throws InterruptedException {
			return CompletelyFairExecutorService.this.awaitTermination(timeout, unit);
		}

		@Override
		public void execute(final Runnable command) {
			if (isShutdown()) {
				throw new RejectedExecutionException(
						"Task " + command.toString() + " rejected from " + CompletelyFairExecutorService.this.toString() + ". Thread pool is shutdown.");
			}

			queue.offer(wrapCommand(command));
		}

		@Override
		public void newContext() {
			throw new UnsupportedOperationException();
		}

		@Override
		public void clearContext() {
			throw new UnsupportedOperationException();
		}

		@Override
		public long getCurrentContextId() {
			return queue.getCurrentContextId();
		}

		@Override
		public FairExecutorService wrapperWithNewContext() {
			return CompletelyFairExecutorService.this.wrapperWithNewContext();
		}

		@Override
		public FairExecutorService wrapperWithCurrentContext() {
			return this;
		}

		@Override
		public CompletableFuture<?> submit(final Runnable task) {
			return FairExecutorService.super.submit(task);
		}

		@Override
		public <T> CompletableFuture<T> submit(final Runnable task, final T result) {
			return FairExecutorService.super.submit(task, result);
		}

		@Override
		public <T> CompletableFuture<T> submit(final Callable<T> task) {
			return FairExecutorService.super.submit(task);
		}

	}

	protected ThreadFactory getFactory() {
		return this.threadFactory;
	}

	protected int getOriginalCorePoolSize() {
		return this.corePoolSize;
	}

	protected int getOriginalMaxPoolSize() {
		return this.maxPoolSize;
	}

	protected long getKeepAliveAmount() {
		return this.keepAliveTime;
	}

	protected TimeUnit getKeepAliveUnit() {
		return this.timeUnit;
	}

	public int getMaxLeasesPerThread() {
		return this.maxLeasesPerThread;
	}

	public int getLeasesPerThread() {
		return this.leasesPerThread;
	}

	public static final class Builder {

		private int corePoolSize = -1;
		private int maxPoolSize = -1;
		private int leasesPerThread = -1;
		private int maxLeasesPerThread = -1;
		private ThreadFactory threadFactory = null;
		private long keepAliveTime = Long.MAX_VALUE;
		private TimeUnit timeUnit = TimeUnit.NANOSECONDS;

		private final AtomicBoolean used = new AtomicBoolean(false);

		private Builder() {
		}

		public void setCorePoolSize(final int corePoolSize) {
			this.corePoolSize = corePoolSize;
		}

		public Builder withCorePoolSize(final int corePoolSize) {
			setCorePoolSize(corePoolSize);
			return this;
		}

		public void setMaxPoolSize(final int maxPoolSize) {
			this.maxPoolSize = maxPoolSize;
		}

		public Builder withMaxPoolSize(final int maxPoolSize) {
			setMaxPoolSize(maxPoolSize);
			return this;
		}

		public void setLeasesPerThread(final int leasesPerThread) {
			this.leasesPerThread = leasesPerThread;
		}

		public Builder withLeasesPerThread(final int leasesPerThread) {
			setLeasesPerThread(leasesPerThread);
			return this;
		}

		public void setMaxLeasesPerThread(final int maxLeasesPerThread) {
			this.maxLeasesPerThread = maxLeasesPerThread;
		}

		public Builder withMaxLeasesPerThread(final int maxLeasesPerThread) {
			setMaxLeasesPerThread(maxLeasesPerThread);
			return this;
		}

		public void setKeepAliveTime(final long keepAliveTime) {
			this.keepAliveTime = keepAliveTime;
		}

		public void setKeepAliveTime(final long keepAliveTime, final TimeUnit unit) {
			this.keepAliveTime = keepAliveTime;
			this.timeUnit = unit;
		}

		public Builder withKeepAliveTime(final long keepAliveTime) {
			setKeepAliveTime(keepAliveTime);
			return this;
		}

		public Builder withKeepAliveTime(final long keepAliveTime, TimeUnit unit) {
			setKeepAliveTime(keepAliveTime, unit);
			return this;
		}

		public void setTimeUnit(final TimeUnit timeUnit) {
			this.timeUnit = timeUnit;
		}

		public Builder withTimeUnit(final TimeUnit timeUnit) {
			this.timeUnit = timeUnit;
			return this;
		}

		public void setThreadFactory(final ThreadFactory threadFactory) {
			this.threadFactory = threadFactory;
		}

		public Builder withThreadFactory(final ThreadFactory threadFactory) {
			setThreadFactory(threadFactory);
			return this;
		}

		public void setThreadNameFormat(final String threadNameFormat) {
			if (threadFactory != null) {
				throw new IllegalStateException("Can't set thread name format when specifying a ThreadFactory");
			}

			this.threadFactory = new NamedThreadFactory(threadNameFormat);
		}

		public Builder withThreadNameFormat(final String threadNameFormat) {
			setThreadNameFormat(threadNameFormat);
			return this;
		}

		public CompletelyFairExecutorService build() {
			if (maxPoolSize == -1) {
				maxPoolSize = corePoolSize;
			}

			if (leasesPerThread == -1) {
				leasesPerThread = corePoolSize;
			}
			if (leasesPerThread <= 0) {
				throw new IllegalArgumentException("leasesPerThread");
			}

			if (maxLeasesPerThread == -1) {
				maxLeasesPerThread = leasesPerThread;
			}
			if (maxLeasesPerThread < leasesPerThread) {
				throw new IllegalArgumentException("maxLeasesPerThread");
			}

			if (!used.compareAndSet(false, true)) {
				throw new IllegalStateException("used");
			}
			if (threadFactory == null) {
				threadFactory = Executors.defaultThreadFactory();
			}

			return new CompletelyFairExecutorService(corePoolSize, maxPoolSize, leasesPerThread, maxLeasesPerThread, keepAliveTime, timeUnit, threadFactory);
		}

	}

	/* EMAPI */

	private static final ExecutorFeatureExtensions[] SUPPORTED = new ExecutorFeatureExtensions[]
		{ ExecutorFeatureExtensions.EEX_query_batching_supported, ExecutorFeatureExtensions.EEX_query_slots_supported,
				ExecutorFeatureExtensions.EEX_compat_native_concurrency_api, ExecutorFeatureExtensions.EEX_executor_low_priority_threads,
				ExecutorFeatureExtensions.EEX_query_debug, };

	@Override
	public boolean supportsFeature(final ExecutorFeatureExtensions feature) {
		return ArrayUtils.contains(SUPPORTED, feature);
	}

	@Override
	public ExecutorFeatureExtensions[] supported() {
		return SUPPORTED;
	}

	/* */

	@Override
	public boolean executorSupportsBatching() {
		return false;
	}

	/* EEX_executor_low_priority_threads */

	@Override
	public boolean supportsLPT() {
		return false;
	}

	@Override
	public boolean isLPTActive() {
		return false;
	}

	@Override
	public void setLPTActive(final boolean state) {
		this.unsupported(ExecutorFeatureExtensions.EEX_executor_low_priority_threads);
	}

	@Override
	public int[] getLPTs() {
		this.unsupported(ExecutorFeatureExtensions.EEX_executor_low_priority_threads);
		return null;
	}

	@Override
	public void submitLP(final Runnable command) {
		TEConditions.validRunnable(command);
		this.submit(command);
		return;
	}

	@Override
	public void submitLP(final Executable<?> r) {
		TEConditions.validExecutable(r);
		if (r instanceof Runnable) {
			this.submit((Runnable) r);
		} else {
			this.submit(() -> {
				r.run_st();
			});
		}
	}

	/* EEX_query_debug */

	@Override
	public Object debug_query(final EEXDebugQueryType type) {
		switch (type) {
			case EEX_DEBUG_TOTAL_THREAD_COUNT:
				return this.maxPoolSize;
			case EEX_DEBUG_ACTIVE_THREAD_COUNT:
				return this.getPoolSize();
			/* */
			default:
				return null;
		}
	}

}